#ifndef _WIN32
#include "CNetServiceImpl.h"
#include "../threadpool.h"
#include "../coroutine.h"
#include <sys/epoll.h>
#include <unistd.h>
#include <signal.h>
#include <sys/resource.h>

namespace TB
{
	thread_local bool CNetServiceImpl::m_bInitedEnv = false;

	CNetServiceImpl::CNetServiceImpl()
	{
		m_nEpollFD = epoll_create(8192);
		QWERASSERT(m_nEpollFD >= 0, );
	}

	CNetServiceImpl::~CNetServiceImpl()
	{
		Close();
	}

	bool CNetServiceImpl::Init(const std::weak_ptr<CCoroutine>& poCoroutine, const std::function<void(INT32)>& rcbOnPollError, const std::weak_ptr<CNetServiceImpl>& poSelf)
	{
		if (!m_poCoroutine.expired())
			return false;
		auto poSharedPtr = poCoroutine.lock();
		if (!poSharedPtr)
			return false;
		m_poCoroutine = poCoroutine;
		m_cbOnPollError = rcbOnPollError;
		m_poSelf = poSelf;
		poSharedPtr->Post([poSelf]()
		{
			auto poImpl = poSelf.lock();
			if (poImpl == nullptr)
				return;
			poImpl->Poll();
		});
		return true;
	}

	void CNetServiceImpl::Poll()
	{
		if (!m_bInitedEnv) 
		{
			sigset_t signal_mask;
			sigemptyset(&signal_mask);
			sigaddset(&signal_mask, SIGPIPE);
			pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);

			rlimit sLimit = {};
			if (0 == getrlimit(RLIMIT_NOFILE, &sLimit))
			{
				sLimit.rlim_cur = sLimit.rlim_max - 1;
				setrlimit(RLIMIT_NOFILE, &sLimit);
			}
			m_bInitedEnv = true;
		}

		INT32 dwCount = 0;
		while (true)
		{
			dwCount = _Poll();
			if (dwCount == 0)
				break;
			if (dwCount < 0)
			{
				auto nErrNo = errno;
				if (nErrNo == EINTR)
					continue;
				if (m_qwPollEventCallbackID != 0)
				{
					CThreadPool::DelFD(m_nEpollFD);
					CThreadPool::ReleasePollEventCallback(m_qwPollEventCallbackID);
					m_qwPollEventCallbackID = 0;
				}
				m_cbOnPollError(nErrNo);
				return;
			}
		}
		if (m_qwPollEventCallbackID == 0)
		{
			std::weak_ptr<CNetServiceImpl> poSelf = m_poSelf;
			m_qwPollEventCallbackID = CThreadPool::CreatePollEventCallback([poSelf](UINT32 dwReturnEvents)
			{
				auto poImpl = poSelf.lock();
				if (!poImpl)
					return;
				auto poCoroutine = poImpl->GetCoroutine().lock();
				if (!poCoroutine)
					return;
				poCoroutine->Post([poSelf]()
				{
					auto poImpl = poSelf.lock();
					if (poImpl == nullptr)
						return;
					poImpl->Poll();
				});
			});
			CThreadPool::AddFD(m_nEpollFD, EPOLLIN | EPOLLONESHOT, m_qwPollEventCallbackID);
		}
		else
		{
			CThreadPool::ModFD(m_nEpollFD, EPOLLIN | EPOLLONESHOT, m_qwPollEventCallbackID);
		}
	}

	INT32 CNetServiceImpl::_Poll()
	{
		int nGot = epoll_wait(m_nEpollFD, m_aEpollEvents, MAX_EPOLL_EVENT_COUNT, 0);
		if (0 >= nGot)
			return nGot;
		for (int j = 0; j != nGot; ++j)
		{
			epoll_event& ee = m_aEpollEvents[j];
			auto it = m_oPollCallbacks.find(ee.data.fd);
			if (it == m_oPollCallbacks.end())
				continue;
			DOnNetPollEvent& rcbCallback = it->second;
			if (rcbCallback)
				rcbCallback(ee.data.fd, ee.events);
		}
		return nGot;
	}

	bool CNetServiceImpl::AddFD(int dwFD, UINT32 dwEvents, const DOnNetPollEvent& cbOnPollEvent)
	{
		if (!m_oPollCallbacks.insert({ dwFD, cbOnPollEvent }).second)
			return false;
		epoll_event ev;
		ev.data.fd = dwFD;
		ev.events = dwEvents;
		if (0 > epoll_ctl(m_nEpollFD, EPOLL_CTL_ADD, dwFD, &ev))
			return false;
		return true;
	}

	bool CNetServiceImpl::ModFD(int dwFD, UINT32 dwEvents)
	{
		epoll_event ev;
		ev.data.fd = dwFD;
		ev.events = dwEvents;
		if (0 > epoll_ctl(m_nEpollFD, EPOLL_CTL_MOD, dwFD, &ev))
			return false;
		return true;
	}

	bool CNetServiceImpl::DelFD(int dwFD)
	{
		if (0 > epoll_ctl(m_nEpollFD, EPOLL_CTL_DEL, dwFD, NULL))
			return false;
		return true;
	}

	void CNetServiceImpl::Close()
	{
		if (m_qwPollEventCallbackID != 0)
		{
			if (m_nEpollFD >= 0)
				CThreadPool::DelFD(m_nEpollFD);
			CThreadPool::ReleasePollEventCallback(m_qwPollEventCallbackID);
			m_qwPollEventCallbackID = 0;
		}
		if (m_nEpollFD >= 0)
		{
			close(m_nEpollFD);
			m_nEpollFD = -1;
		}
	}
}
#endif